package cn.tpshion.pulsar.support;

import org.apache.pulsar.client.api.Producer;
import org.apache.pulsar.client.api.PulsarClient;
import org.springframework.beans.factory.FactoryBean;

import java.util.concurrent.TimeUnit;

public class ProducerFactoryBean implements FactoryBean<Producer> {

    private PulsarClient pulsarClient;

    private PulsarProducer producer;

    public ProducerFactoryBean(PulsarClient pulsarClient, PulsarProducer producer) {
        this.pulsarClient = pulsarClient;
        this.producer = producer;
    }

    @Override
    public Producer getObject() throws Exception {
        return pulsarClient.newProducer(producer.getSchema())
                .producerName(producer.getProducerName())
                .topic(producer.buildTopic())
                .enableBatching(producer.isBatchingEnabled())
                .sendTimeout(producer.getSendTimeoutMs(), TimeUnit.MILLISECONDS)
                .blockIfQueueFull(producer.isBlockIfQueueFull())
                .maxPendingMessages(producer.getMaxPendingMessages())
                .batchingMaxMessages(producer.getBatchingMaxMessages())
                .enableChunking(producer.isChunkingEnabled())
                .create();
    }

    @Override
    public Class<?> getObjectType() {
        return Producer.class;
    }

    @Override
    public boolean isSingleton() {
        return true;
    }

}
